package com.alibaba.citrus.cr.mn.order.center.facade.kafka;

import com.alibaba.cz.base.tool.log.Log;
import com.alibaba.lattice2.epoch.sdk.util.SpringBeanContextUtil;
import org.apache.commons.lang3.StringUtils;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.Objects;
import java.util.UUID;

/**
 * Chen.mu
 * 2023/4/23
 **/
public class KafkaUtils {
    private static final Log log = Log.getLogger(KafkaUtils.class);
    private static KafkaTemplate kafkaTemplate = SpringBeanContextUtil.getBean(KafkaTemplate.class);

    public static SendResult<Integer, String> send(String topic, String data) {
        return send(topic, UUID.randomUUID().toString(), data);
    }

    public static SendResult<Integer, String> send(String topic, String key, String data) {
        ListenableFuture<SendResult<Integer, String>> future;
        if (Objects.isNull(key)) {
            future = kafkaTemplate.send(topic, data);
        } else {
            future = kafkaTemplate.send(topic, key, data);
        }
        try {
            log.info("KafkaUtils#send topic: {}", topic);
            SendResult<Integer, String> result = future.get();
            log.info("KafkaUtils#send result: {}", result);
            return result;
        } catch (Exception e) {
            log.error("KafkaUtils#send error", e);
        }
        return null;
    }

    /**
     * 生成消息主键
     *
     * @param fromSys          来源系统
     * @param businessType     业态
     * @param businessUnitCode 业务单元编码
     * @param eventCode        事件编码
     * @return
     */
    public static String generateMessageKey(String fromSys, String businessType, String businessUnitCode, String eventCode) {
        String splitKey = "-";
        StringBuilder messageKey = new StringBuilder();
        messageKey.append(fromSys).append(splitKey)
                .append(StringUtils.isBlank(businessType) ? "99" : businessType).append(splitKey)
                .append(StringUtils.isBlank(businessUnitCode) ? "XX" : businessUnitCode).append(splitKey)
                .append(eventCode).append(splitKey)
                .append(UUID.randomUUID().toString().replaceAll("-", ""));
        return messageKey.toString();
    }
}

